package study.zookeeper.support.leader;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.CancelLeadershipException;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.state.ConnectionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 与LeaderLatch， 通过LeaderSelectorListener可以对领导权进行控制，
 * 在适当的时候释放领导权，这样每个节点都有可能获得领导权。 而LeaderLatch一根筋到死，
 * 除非调用close方法，否则它不会释放领导权。
 * 异常处理：
 * LeaderSelectorListener类继承ConnectionStateListener.LeaderSelector必须小心连接状态的 改变
 * 如果实例成为leader, 它应该相应SUSPENDED 或 LOST. 当 SUSPENDED 状态出现时
 * 实例必须假定在重新连接成功之前它可能不再是leader了。 如果LOST状态出现， 实例不再是leader，takeLeadership方法返回
 * 区别：
 * leaderlatch需要调用close方法才能释放主导权，并且不能重新获得
 * leaderselector当执行完takeleadership方法后自动释放主导权，并且可以设置autorequeue重新再获取领导权
 * 实现方式不同leaderselector使用分布式锁InterProcessMutex实现
 */
@Slf4j
@Component
public class LeaderCommandLineRunner extends LeaderSelectorListenerAdapter implements CommandLineRunner {

    /* leader 次数 */
    public AtomicInteger leaderCount = new AtomicInteger(1);
    /*  */
    private String path = "/leader_selector";
    /*  */
    private LeaderSelector selector;

    @Autowired
    private CuratorFramework client;

    @Override
    public void stateChanged(CuratorFramework client, ConnectionState newState) {
        if (newState == ConnectionState.SUSPENDED) {
            throw new CancelLeadershipException();
        }
    }

    /**
     * 当实例被选为leader之后，调用takeLeadership方法进行业务逻辑处理，处理完成即释放领导权。
     * 其中autoRequeue()方法的调用确保此实例在释放领导权后还可能获得领导权。
     */
    @Override
    public void takeLeadership(CuratorFramework client) throws Exception {
        log.info("共成为leader的次数：{} 次", leaderCount.getAndIncrement());
//        selector.autoRequeue();
    }

    @Override
    public void run(String... args) throws Exception {
        selector = new LeaderSelector(client, path, this);
        selector.start();
    }
}
